package cn.codeartist.springboot.component.rocketmq.listener;

import cn.codeartist.springboot.component.rocketmq.annotation.RocketMQConsumer;
import cn.codeartist.springboot.component.rocketmq.annotation.RocketMQMessageListener;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * RocketMQ注册消费者以及绑定监听器
 *
 * @author 艾江南
 * @date 2020/8/12
 */
public class RocketMQConsumerListener implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {

    private final static Logger logger = LoggerFactory.getLogger(RocketMQConsumerListener.class);

    private final List<DefaultMQPushConsumer> mqPushConsumerList = new ArrayList<>();

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        ApplicationContext applicationContext = event.getApplicationContext();
        // 防止重复绑定
        if (applicationContext.getParent() != null) {
            return;
        }
        String beanName = null;
        try {
            Map<String, Object> beans = applicationContext.getBeansWithAnnotation(RocketMQConsumer.class);
            for (Map.Entry<String, Object> entry : beans.entrySet()) {
                beanName = entry.getKey();
                Object object = entry.getValue();
                for (Method method : object.getClass().getMethods()) {
                    RocketMQMessageListener listener = AnnotationUtils.findAnnotation(method, RocketMQMessageListener.class);
                    if (listener == null) {
                        continue;
                    }
                    // 获取消费者Bean实例
                    DefaultMQPushConsumer mqPushConsumer = applicationContext.getBean(listener.consumer(), DefaultMQPushConsumer.class);
                    logger.info("RocketMQ consumer register listener with bean:{}, group:{}, topic:{}, tag:{}", listener.consumer(), listener.group(),
                            listener.topic(), listener.tag());
                    // 是否使用注解配置Group、Topic、Tag
                    if (!StringUtils.isEmpty(listener.group())) {
                        mqPushConsumer.setConsumerGroup(listener.group());
                    }
                    if (!(StringUtils.isEmpty(listener.topic()) || StringUtils.isEmpty(listener.tag()))) {
                        mqPushConsumer.subscribe(listener.topic(), listener.tag());
                    }
                    // 注册监听器
                    mqPushConsumer.setMessageListener(getMessageListener(object, method));
                    mqPushConsumerList.add(mqPushConsumer);
                }
            }
            // 启动所有消费者
            start();
        } catch (Exception e) {
            logger.error("RocketMQ consumer register listener error. beanName:{}", beanName, e);
        }
    }

    @Override
    public void destroy() {
        for (DefaultMQPushConsumer consumer : mqPushConsumerList) {
            if (consumer != null) {
                consumer.shutdown();
                logger.info("RocketMQ consumer destroy with {}", consumer);
            }
        }
    }

    private void start() throws MQClientException {
        for (DefaultMQPushConsumer consumer : mqPushConsumerList) {
            if (consumer != null) {
                long start = System.currentTimeMillis();
                consumer.start();
                long end = System.currentTimeMillis();
                logger.info("RocketMQ consumer start with {} cost {}ms", consumer, end - start);
            }
        }
    }

    /**
     * 将方法绑定到RocketMQ监听器中
     *
     * @param object RocketMQ消费者对象
     * @param method RocketMQ监听方法
     * @return RocketMQ监听接口
     */
    private MessageListenerConcurrently getMessageListener(Object object, Method method) {
        Parameter[] parameters = method.getParameters();
        // 暂时只支持0个或1个参数的方法
        if (parameters.length > 1) {
            throw new IllegalArgumentException("RocketMQ consumer init error by method args number is <=1.");
        }
        return (msgs, context) -> {
            if (CollectionUtils.isEmpty(msgs)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            try {
                for (MessageExt messageExt : msgs) {
                    // 自动JSON解析
                    String msg = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                    Object[] params = new Object[parameters.length];
                    for (int i = 0; i < parameters.length; i++) {
                        Class<?> clazz = parameters[i].getType();
                        params[i] = clazz == String.class ? msg : JSON.parseObject(msg, clazz);
                    }
                    logger.info("RocketMQ consumer handle message by {}#{}, message:{}", object.getClass().getSimpleName(), method.getName(), msg);
                    method.invoke(object, params);
                }
            } catch (JSONException e) {
                logger.error("RocketMQ consumer handle message exception by {}#{}", object.getClass().getSimpleName(), method.getName(), e);
            } catch (Exception e) {
                logger.error("RocketMQ consumer handle message exception by {}#{}", object.getClass().getSimpleName(), method.getName(), e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        };
    }
}
